package com.qudian.pay.monitor.logcollect.logback;

import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.qudian.pay.monitor.logcollect.logback.layout.LayoutMessageEncoder;
import com.qudian.pay.mq.SendMessage;
import com.qudian.pay.mq.producer.impl.MQProducerImpl;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;

public final class MQAppender extends AppenderBase<ILoggingEvent> {
	
	private static final Logger logger = LoggerFactory.getLogger(MQAppender.class);
	
	protected LayoutMessageEncoder<ILoggingEvent> encoder;
	
	private final ConcurrentLinkedQueue<ILoggingEvent> queue = Queues.newConcurrentLinkedQueue();
	
	/**
	 * @Fields queueThreshold : 队列大小
	 */
	private int queueThreshold = 1;
	
	private String namesrvAddr;
	
	private String instanceName;
	
	private String topic;
	
	private MQProducerImpl mqProducer;
	
	@Override
	public void start() {
		try {
			if (this.encoder == null) {
				logger.warn("no encoder definite");
			} else {
				encoder.start();
			}
			super.start();
			mqProducer = new MQProducerImpl();
			mqProducer.setNamesrvAddr(namesrvAddr);
			mqProducer.setInstanceName(instanceName);
			mqProducer.setGroup("finance-group");
			mqProducer.init();
		} catch (Exception e) {
			logger.error("MQAppender初始化失败，启动rocketMQ异常", e);
		}
	}
	
	@Override
	public void stop() {
		this.sendMQLog(true);
		super.stop();
		mqProducer.destroy();
		logger.info("Stopping MQAppender...");
	}
	
	@Override
	protected void append(ILoggingEvent event) {
		queue.add(event);
		sendMQLog(false);
	}
	
	private void sendMQLog(boolean flag) {
		// 超出队列设置阈值则进行发送
		if (flag || queue.size() >= queueThreshold) {
			try {
				ILoggingEvent loggingEvent = null;
				SendMessage msg = new SendMessage();
				Map<String, Object> bizValue = Maps.newHashMap();
				while ((loggingEvent = queue.poll()) != null) {
					String formattedMessage = loggingEvent.getFormattedMessage();
					if(encoder != null) {
						formattedMessage = encoder.doEncode(loggingEvent);
					}
					bizValue.put("msg", formattedMessage);
					msg.setBizValue(bizValue);
					// 异步发送
					mqProducer.sendMessage(topic, topic, msg, new SendCallback() {
						@Override
						public void onSuccess(SendResult sendResult) {}
						@Override
						public void onException(Throwable e) {
							logger.warn("rocketMQ发送失败",e);
						}
					});
				}
			} catch (Exception e) {
				logger.error("MQAppender: Exception sending message:", e);
			}
		}
	}
	
	public void setTopic(String topic) {
		this.topic = topic;
	}
	
	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}
	
	public void setInstanceName(String instanceName) {
		this.instanceName = instanceName;
	}
	
	public void setEncoder(LayoutMessageEncoder<ILoggingEvent> encoder) {
		this.encoder = encoder;
	}
}